/**
 * Copyright (C) 2016 Eshore Corporation
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.eshore.thrpc.config;

import java.lang.reflect.Constructor;

import org.apache.commons.lang3.StringUtils;
import org.apache.thrift.TProcessor;

import com.eshore.thrpc.RpcException;
import com.eshore.thrpc.common.NetUtils;
import com.eshore.thrpc.common.ServerNode;
import com.eshore.thrpc.proxy.DynamicServiceHandler;
import com.eshore.thrpc.server.IServer;
import com.eshore.thrpc.server.thrift.ThriftServerFactory;

/**
 * 服务提供者配置
 * <p>
 * 
 * @author : zhouzhixiang
 * @date : 2016-11-19
 */
public class ServerConfig implements IConfigCheck {
	/** 服务名 */
    private String name;

    /** 服务负责人 */
    private String owner;

    /** 端口，默认是19090 */
    private int port = 19090;

    /** 协议 */
    private String protocol = "thrift";

    /** 服务ip地址，可以为空，为空时从网卡获取 */
    private String ip;

    /** 服务实现类 */
    private Object ref;

    /** 服务名(全称)：命名空间$服务名简称 */
    private String service;

    /** 最大工作线程数，默认为100 */
    private int maxWorkerThreads = 100;

    /** 最小工作线程数 ,默认为10 */
    private int minWorkerThreads = 10;

    /** {@link IServer} */
    private IServer server;
    
    /** 服务端阻塞方式 */
    private boolean isNonBlock = true;
    
    /**
     *
     * ISMP+不是CPU密集型系统，调整稍微大点，默认150
     * 工作线程数
     * 负责协议的解析 对象序列化 业务逻辑的执行 响应写入到ByteBuffer 工作线程为0 就直接在selector线程上执行  
     */
    private int workerThreads = 150;
    
    /**
     * 每个selector线程中请求的最大个数，这些请求还没有注册读取事件  
     */
    private int acceptQueueSizePerThread = 10000;
    
    /**
     * selector线程数
     * selector线程负责1.事件注册 2.读写IO轮询 3.将数据读取或者写入到transport
     */
    private int selectorThreads = 2;
    
    
    public void export() throws ClassNotFoundException, RpcException {
    	//参数检查
        check();

        ServerNode serverNode = genServerNode();

        // 创建服务
        IServer server = createServer(serverNode);
        server.start();
        addShutdownHook(server);
        if (server.isStarted()) {
            this.server = server;
        } else {
            server.stop();
        }
    }
    

    /**
     * 创建服务
     * <p>
     * 
     * @param serverNode
     * @param rpcMonitor
     * @return {@link IServer}
     * @throws ClassNotFoundException
     * @throws RpcException 
     */
    protected IServer createServer(ServerNode serverNode) throws ClassNotFoundException, RpcException {
        IServer server = null;
        TProcessor processor = reflectProcessor(serverNode);
        if(isNonBlock){
            server = ThriftServerFactory.createNonBlockingServer(processor, serverNode, selectorThreads, workerThreads, acceptQueueSizePerThread);
        }else{
        	server = ThriftServerFactory.createBlockingServer(processor, serverNode, maxWorkerThreads, minWorkerThreads);
        }
        
        return server;
    }

    /**
     * 反射TProcessor
     * <p>
     * 
     * @param rpcMonitor
     * @param serverNode
     * @return TProcessor
     * @throws RpcException 
     */
    @SuppressWarnings("rawtypes")
    protected TProcessor reflectProcessor(ServerNode serverNode) throws RpcException {
        Class serviceClass = getRef().getClass();
        ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
        Class<?>[] interfaces = serviceClass.getInterfaces();
        if (interfaces.length == 0) {
            throw new RpcException("Service class should implements Iface!");
        }

        // 反射,load "Processor";
        TProcessor processor = null;
        for (Class clazz : interfaces) {
            String cname = clazz.getSimpleName();
            if (!cname.equals("Iface")) {
                continue;
            }
            String pname = clazz.getEnclosingClass().getName() + "$Processor";
            try {
                Class<?> pclass = classLoader.loadClass(pname);
                Constructor constructor = pclass.getConstructor(clazz);
                processor = (TProcessor) constructor.newInstance(getRef());
//                processor = (TProcessor) constructor.newInstance(getProxy(classLoader, clazz, getRef(), serverNode));
            } catch (Exception e) {
                throw new RpcException("Refact error,please check your thift gen class!", e.getCause());
            }
        }

        if (processor == null) {
            throw new RpcException("Service class should implements $Iface!");
        }
        return processor;
    }
    

    /**
     * 获取处理类代理,暂时未使用
     * <p>
     * 
     * @param classLoader
     * @param interfaces
     * @param object
     * @param rpcMonitor
     * @param serverNode
     * @return 处理类代理
     * @throws ClassNotFoundException
     */
    @SuppressWarnings("unused")
	private Object getProxy(ClassLoader classLoader, Class<?> interfaces, Object object, ServerNode serverNode) throws ClassNotFoundException {
        DynamicServiceHandler dynamicServiceHandler = new DynamicServiceHandler();
        return dynamicServiceHandler.bind(classLoader, interfaces, object, serverNode);
    }

    /**
     * 反射iface
     * <p>
     * 
     * @return {@link Responder}
     * @throws RpcException 
     */
    @SuppressWarnings("rawtypes")
    protected Class reflectProtocolClass() throws RpcException {
        Class serviceClass = getRef().getClass();
        Class<?>[] interfaces = serviceClass.getInterfaces();
        if (interfaces.length == 0) {
            throw new RpcException("Service class should implements avro's interface!");
        }

        // 生成Responder
        for (Class clazz : interfaces) {
            try {
                clazz.getDeclaredField("PROTOCOL").get(null);
                return clazz;
                // return new SpecificResponder(clazz, ref);
            } catch (Exception e) {
            }
        }
        throw new RpcException("Service class should implements avro's interface!");
    }

    /**
     * 生成 {@link ServerNode}
     * <p>
     * 
     * @return {@link ServerNode}
     * @throws RpcException 
     */
    protected ServerNode genServerNode() throws RpcException {
        String ip = null;
        if (StringUtils.isNotEmpty(getIp())) {
            ip = getIp();
        } else {
            ip = NetUtils.getLocalHost();
        }
        
        if(StringUtils.isEmpty(name)){
        	name="THRPC SERVER";
        }
        
        
        if (ip == null) {
            throw new RpcException("Can't find server ip!");
        }
        
        
        return new ServerNode(ip, getPort(),name,isNonBlock);
    }

    @Override
    public void check() throws RpcException {
        if (StringUtils.isEmpty(service)) {
            throw new RpcException(RpcException.CONFIG_EXCEPTION, "The params 'service' cannot empty!");
        }
    }

    /**
     * 销毁资源<br/>
     * 包括：释放注册中心连接、停止服务。
     * <p>
     */
    public void destory() {
        if (server != null) {
            server.stop();
        }
    }
    
    /**
     * 添加关闭钩子
     * <p>
     * 
     * @param registry
     * @param server
     */
    protected void addShutdownHook(final IServer server) {
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            @Override
            public void run() {
                if (server != null) {
                    server.stop();
                }
            }
        }));
    }

    /**
     * getter method
     * 
     * @see ServerConfig#name
     * @return the name
     */
    public String getName() {
        return name;
    }

    /**
     * setter method
     * 
     * @see ServerConfig#name
     * @param name
     *            the name to set
     */
    public void setName(String name) {
        this.name = name;
    }

    /**
     * getter method
     * 
     * @see ServerConfig#owner
     * @return the owner
     */
    public String getOwner() {
        return owner;
    }

    /**
     * setter method
     * 
     * @see ServerConfig#owner
     * @param owner
     *            the owner to set
     */
    public void setOwner(String owner) {
        this.owner = owner;
    }

    /**
     * getter method
     * 
     * @see ServerConfig#port
     * @return the port
     */
    public int getPort() {
        return port;
    }

    /**
     * setter method
     * 
     * @see ServerConfig#port
     * @param port
     *            the port to set
     */
    public void setPort(int port) {
        this.port = port;
    }

    /**
     * getter method
     * 
     * @see ServerConfig#protocol
     * @return the protocol
     */
    public String getProtocol() {
        return protocol;
    }

    /**
     * setter method
     * 
     * @see ServerConfig#protocol
     * @param protocol
     *            the protocol to set
     */
    public void setProtocol(String protocol) {
        this.protocol = protocol;
    }

    /**
     * getter method
     * 
     * @see ServerConfig#ip
     * @return the ip
     */
    public String getIp() {
        return ip;
    }

    /**
     * setter method
     * 
     * @see ServerConfig#ip
     * @param ip
     *            the ip to set
     */
    public void setIp(String ip) {
        this.ip = ip;
    }

    /**
     * getter method
     * 
     * @see ServerConfig#ref
     * @return the ref
     */
    public Object getRef() {
        return ref;
    }

    /**
     * setter method
     * 
     * @see ServerConfig#ref
     * @param ref
     *            the ref to set
     */
    public void setRef(Object ref) {
        this.ref = ref;
    }

    /**
     * getter method
     * 
     * @see ServerConfig#service
     * @return the service
     */
    public String getService() {
        return service;
    }

    /**
     * setter method
     * 
     * @see ServerConfig#service
     * @param service
     *            the service to set
     */
    public void setService(String service) {
        this.service = service;
    }
    
    /**
     * getter method
     * 
     * @see ServerConfig#maxWorkerThreads
     * @return the maxWorkerThreads
     */
    public int getMaxWorkerThreads() {
        return maxWorkerThreads;
    }

    public IServer getServer() {
        return server;
    }


    public void setServer(IServer server) {
        this.server = server;
    }


    /**
     * setter method
     * 
     * @see ServerConfig#maxWorkerThreads
     * @param maxWorkerThreads
     *            the maxWorkerThreads to set
     */
    public void setMaxWorkerThreads(int maxWorkerThreads) {
        this.maxWorkerThreads = maxWorkerThreads;
    }

    /**
     * getter method
     * 
     * @see ServerConfig#minWorkerThreads
     * @return the minWorkerThreads
     */
    public int getMinWorkerThreads() {
        return minWorkerThreads;
    }

    /**
     * setter method
     * 
     * @see ServerConfig#minWorkerThreads
     * @param minWorkerThreads
     *            the minWorkerThreads to set
     */
    public void setMinWorkerThreads(int minWorkerThreads) {
        this.minWorkerThreads = minWorkerThreads;
    }

	public boolean isNonBlock() {
		return isNonBlock;
	}

	public void setNonBlock(boolean isNonBlock) {
		this.isNonBlock = isNonBlock;
	}


	public int getWorkerThreads() {
		return workerThreads;
	}


	public void setWorkerThreads(int workerThreads) {
		this.workerThreads = workerThreads;
	}


	public int getAcceptQueueSizePerThread() {
		return acceptQueueSizePerThread;
	}

	public void setAcceptQueueSizePerThread(int acceptQueueSizePerThread) {
		this.acceptQueueSizePerThread = acceptQueueSizePerThread;
	}

	public int getSelectorThreads() {
		return selectorThreads;
	}

	public void setSelectorThreads(int selectorThreads) {
		this.selectorThreads = selectorThreads;
	}
	
}
